package com.zdb.demo.flink

import org.apache.flink.api.common.functions.{AggregateFunction, FilterFunction, MapFunction}
import org.apache.flink.api.java.functions.KeySelector
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.datastream.DataStreamSource
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
import org.apache.flink.streaming.api.scala._
import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, TypeInformation, Types}

// 输入数据样例类
case class UserBehavior( userId: Long, itemId: Long, categoryId: Int, behavior: String, timestamp: Long )

// 中间聚合结果样例类
case class ItemViewCount( itemId: Long, windowEnd: Long, count: Long )

object HotItems {

  def main(args: Array[String]): Unit = {
    val conf = new Configuration() {}
    conf.setInteger("rest.port", 9191)
    conf.setBoolean("local.start-webserver", true)
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    env.setParallelism(1)

    val dstream: DataStreamSource[String] = env.readTextFile("/Users/zhudebin/Documents/iworkspace/flink-demo/hotItemsAnalysis/src/main/resources/UserBehavior.csv")

    dstream.map(new MapFunction[String, UserBehavior] {
      override def map(value: String): UserBehavior = {
        val strs = value.split(",")
        UserBehavior(strs(0).trim.toInt, strs(1).trim.toInt,
          strs(2).trim.toInt, strs(3).trim, strs(4).trim.toLong)
      }
    }).filter(new FilterFunction[UserBehavior] {
      override def filter(value: UserBehavior): Boolean = {
        value.behavior == "pv"
      }
    }).keyBy(new KeySelector[UserBehavior, Long] {
      override def getKey(value: UserBehavior): Long = {
        value.itemId
      }
    }).timeWindow(Time.minutes(60), Time.minutes(5))
      /**
        .aggregate(new AggregateFunction[UserBehavior, Long, Long] {
          override def createAccumulator(): Long = {
            0L
          }

          override def add(value: UserBehavior, accumulator: Long): Long = {
            accumulator + 1
          }

          override def getResult(accumulator: Long): Long = {
            accumulator
          }

          override def merge(a: Long, b: Long): Long = {
            a + b
          }
        },
          new WindowFunction[Long, ItemViewCount, Long, TimeWindow] {
            override def apply(key: Long, window: TimeWindow,
                               input: Iterable[Long],
                               out: Collector[ItemViewCount]): Unit = {

              val ivc = ItemViewCount(key, window.getEnd, 10L)
              out.collect(ivc)
            }
          }, BasicTypeInfo.INT_TYPE_INFO, TypeInformation.of(classOf[ItemViewCount])
        )
      */
      

    env.execute("hot items")

  }
}
